package com.niit.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Date:2025/5/21
 * Author：Ys
 * Description:
 */
object SparkStreaming04_Join {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming04_Join")
    val ssc = new StreamingContext(conf,Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    val lines9999: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    val lines8848: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 8848)

    //只有key相同的情况下才可以合并

    val map9999: DStream[(String, String)] = lines9999.map((_, "9999"))
    val map8848: DStream[(String, String)] = lines8848.map((_, "8848"))

    map9999.join(map8848).print()

    ssc.start()

    ssc.awaitTermination()
  }

}
